package com.yanggu.flink.datastream_api.window

import com.yanggu.flink.datastream_api.pojo.SensorReading
import org.apache.flink.configuration.{Configuration, RestOptions}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow
import org.apache.flink.util.Collector

/**
 * CountWindow
 * 根据窗口中相同key元素的数量来触发执行, 执行时只计算元素 数量达到窗口大小的key对应的结果
 * 注意：CountWindow的window_size指的是相同Key的元素的个数, 不是输入的所有元素的总数
 * 分为滚动窗口和滑动窗口
 */
object CountWindowDemo {

  def main(args: Array[String]): Unit = {
    //创建本地执行环境, 并且拥有WebUi和设置端口
    val config = new Configuration()
    config.setInteger(RestOptions.PORT.key(), 8080)
    val environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config)
    environment.setParallelism(1)

    environment
      .socketTextStream("localhost", 9000)
      .map(value => {
        val arr = value.split(",")
        SensorReading(arr(0), arr(1).toLong * 1000, arr(2).toDouble)
      })
      //根据id进行分区
      .keyBy(_.id)
      //计数滚动窗口, 当元素数量到达窗口的大小时, 就会触发窗口的执行
      //.countWindow(5L)
      //计数滑动窗口size和slide
      //每当某一个key的个数达到2的时候, 触发计算, 计算最近该key最近5个元素
      .countWindow(5L, 2L)
      //这里的Window类型只能写GlobalWindow
      .process(new ProcessWindowFunction[SensorReading, String, String, GlobalWindow] {
        override def process(key: String, context: Context, elements: Iterable[SensorReading], out: Collector[String]): Unit = {
          val result = s"元素的key是: $key, 窗口内元素的个数是: ${elements.size}, 窗口内的数据: ${elements.mkString(",")}"
          out.collect(result)
        }
      })
      .print("result")

    environment.execute(getClass.getSimpleName + " job")

  }

}
